package cn.ac.iie.di.ban.nmpp.handler;

import cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv;
import cn.ac.iie.di.ban.nmpp.tools.MppTools;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import com.alibaba.fastjson.JSONObject;
import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.DefaultRetryPolicy;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import static java.lang.Thread.sleep;
import java.nio.ByteBuffer;
import java.util.*;

import org.apache.log4j.Logger;
import static cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv.UPDATE_FORWARDMQ;

/**
 * Created by jinsheng on 16/9/18.
 *
 */
public class SE2DWWorkerMPPUpdate extends SE2DWWorker {

    private String tableName;
    private String timestampKey = "";
    private String timestampFieldKey;
    private String keySpace;
    long startTime = -1;
//rocketmq Type
    private Map<String, REFieldType> rtn;
    private List<String> updateFields;
    static Map<String, String> ColumnNameType = new HashMap<String, String>();
    private List<ColumnMetadata> columns;
    private Map<String, Type> columnMap = new HashMap<>();

    private final ObjectMapper objectMapper = new ObjectMapper();
    private static org.apache.log4j.Logger logger = Logger.getLogger(SE2DWWorkerMPPUpdate.class.getName());

    private Cluster cluster;
    private Session session;
    private String updatePrimaryKey;
    private PreparedStatement insert_statement = null;
    private PreparedStatement update_statment = null;

    public SE2DWWorkerMPPUpdate(TableSe2DWHandler pTableSe2DWHandler, String pDWType) {
        super(pTableSe2DWHandler, pDWType);
    }

    public List<ColumnMetadata> getColumns() {
        return columns;
    }

    @Override
    public void init() throws Exception {
        tableName = tableSe2DBRule.getTableName();
        if (UPDATE_FORWARDMQ.get(tableName) != null) {
            updateFields = UPDATE_FORWARDMQ.get(tableName).getColumnSet();
            updatePrimaryKey = UPDATE_FORWARDMQ.get(tableName).getQueryid();
        }
        initPStatement();

    }

    private void initConnection() {

        if (cluster != null) {
            return;
        }
        int readTimeOut = 12000;
        Boolean retry = true;
        for (int tryTimes = 0;; tryTimes++) {
            logger.info(tryTimes + "st times try get available connection to " + tableSe2DBRule.getDbType() + " dw");
            String[] dbmeta = tableSe2DBRule.getConnStr().split("[/]");
            if (dbmeta.length != 3) {
                logger.error("unknow dbtype " + tableSe2DBRule.getDbType() + " dw unsuccessfully for " + tableSe2DBRule.getConnStr());
            }
            String userName = tableSe2DBRule.getUserName();
            String password = tableSe2DBRule.getPassword();

            QueryOptions options = new QueryOptions();
            options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
            Cluster.Builder builder = Cluster.builder();
            String dataCenter = dbmeta[2];
            keySpace = dbmeta[1];
//            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
            tableName = tableSe2DBRule.getTableName();
//            String coordinatorAddress = dbmeta[0];
            String [] Address = dbmeta[0].split("[;]");
            Random ran = new Random();
            int num = ran.nextInt(Address.length);
            String coordinatorAddress = Address[num];
            //创建连接
            //需要用户认证
            if (userName.length() > 0) {
                builder
                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                        .withCredentials(userName, password).addContactPoint(coordinatorAddress).withQueryOptions(options);
                // 不需要用户认证{
            } else {
                builder
                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                        .addContactPoint(coordinatorAddress).withQueryOptions(options);
            }
            if (retry == null || retry) {
                this.cluster = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
            } else {
                this.cluster = builder.build();
            }
            break;

        }

    }

    public void initPStatement() {

        if (cluster == null) {
            for (int tryTimes = 0;; tryTimes++) {
                logger.info(tryTimes + "st times try get valid pstmt to table " + tableSe2DBRule.getTableName() + " on " + tableSe2DBRule.getDbType() + " dw");
                initConnection();
                if (cluster == null) {
                    logger.error("fatal error when getting connection to " + tableSe2DBRule.getDbType() + " dw");
                } else {
                    try {
                        logger.info("create pstmt...");
                        long startTime = System.nanoTime();
                        //获取该表的元数据����
                        this.session = this.cluster.connect(keySpace);
                        //处理每一列的类型并保存，用于更新时确定调用哪个set方法
                        logger.info(keySpace + " " + tableName);
                        this.columns = this.cluster.getMetadata().getKeyspace(keySpace).getTable(tableSe2DBRule.getTableName()).getColumns();
//                        columnMap = columns
//                                .parallelStream()
//                                .collect(Collectors.toMap(ColumnMetadata::getName, col -> resolveType(col.getType())));
                        for (ColumnMetadata name : columns) {
                            columnMap.put(name.getName(), resolveType(name.getType()));
                        }
                        logger.info(columnMap.toString());
                        long endTime = System.nanoTime();
                        logger.info("create pstmt successfully using " + (endTime - startTime) / 1000000L + " ms");
                        logger.info("checkmongodb count and " + tableSe2DBRule.getTableName() + " count is   ");
                        if (update_statment == null) {
                            // 预编译update语句
                            String sql = "update " + this.tableName + " set "
                                    + getUpdateStr() + " where " + updatePrimaryKey
                                    + "=:key if exists";// IF EXISTS";
                            logger.info("update sql is  " + sql);
                            update_statment = session.prepare(sql);
                        }
                        break;
                    } catch (Exception ex) {
                        logger.error("create pstmt to " + tableSe2DBRule.getTableName() + " unsuccessfully for " + ex.getMessage(), ex);
                        try {
                            if (session != null) {
                                session.close();
                            }
                        } catch (Exception ex2) {
                        } finally {
                            session = null;
                        }
                    }
                }
                try {
                    Thread.sleep(2000);
                } catch (Exception ex) {
                }
            }
        }
    }

    public void quserykey(List<HashMap> lists) {
        
        for (HashMap m : lists) {
//session.execute(stmnt)
            
        }

    }

    @Override
    public void loadData0() {
        try {
            if (docRecordSet.size() < 1) {
                logger.warn("no record load to table " + tableSe2DBRule.getTableName() + " in " + dwType);
                return;
            }
            int validRecordNum = 0;
            int tryTimes = 0;
            boolean failed = false;
            long endTime = -1;
            for (tryTimes = 0; tryTimes < 10; tryTimes++) {
                logger.info(tryTimes + "st try to write " + docRecordSet.size() + " records to " + tableSe2DBRule.getTableName() + "...");
                initPStatement();
                logger.info(tryTimes + "st set parameters...");
                String columnName = "";
                String rocketMqColumnName = "";
                ArrayList<Map<String, Object>> DataR = new ArrayList<Map<String, Object>>();
                if (rtn == null) {
                    rtn = RuntimeEnv.ROCKETMQ_DOC_DESC.get(tableName);
                }
                startTime = System.nanoTime();
                for (HashMap docRecord : docRecordSet) {
                    Map<String, Object> newSingleData = new HashMap<String, Object>();
                    for (Map.Entry<String, Type> entry : columnMap.entrySet()) {
                        columnName = entry.getKey();
                        rocketMqColumnName = GetColumnName(columnName);
                        if (docRecord.get(rocketMqColumnName) != null) {
                            Object value = docRecord.get(rocketMqColumnName);
                            if (columnName.equals("it") || rocketMqColumnName.equals("m_insert_time")) {
                                newSingleData.put(columnName, System.currentTimeMillis() / 1000);
                            } else {
                                if (rtn.get(rocketMqColumnName) == REFieldType.Struct) {
                                    HashMap n = (HashMap) value;
                                    String json = JSONObject.toJSONString(n);
                                    newSingleData.put(columnName, json);
                                } else if (rtn.get(rocketMqColumnName) == REFieldType.Structs) {
                                    Map[] m = (Map[]) value;
                                    String jsons = MppTools.maps2json(m);
                                    newSingleData.put(columnName, jsons);
                                } else {
                                    Object s = resolveType(entry.getValue(), value);
                                    newSingleData.put(columnName, s);
                                }
                            }
                        } else {
                            if (validRecordNum == 0) {
                                logger.error("cannot getName from rocketmq and set columnName null" + columnName + "  : " + rocketMqColumnName);
                            }
                            switch (entry.getValue()) {
                                case STRING:
                                    newSingleData.put(columnName, "");
                                    break;
                                case INT:
                                    newSingleData.put(columnName, 0);
                                    break;
                                case BIGINT:
                                    newSingleData.put(columnName, 0l);
                                    break;
                                case LIST:
                                    String[] ss = new String[]{};
                                    newSingleData.put(columnName, ss);
                                    break;
                                default:
                                    logger.info("unknow type " + entry.getValue() + " and column Name is " + columnName);
                            }
                        }
                    }
                    DataR.add(newSingleData);
                    validRecordNum++;
                }
                logger.info("data begin loader newmpp and size is " + DataR.size());
                try {
                    while (!updateData(DataR)) {
                        logger.error("insert error  and sleep" + 500 + "ms");
                        sleep(500);
                        close();
                        initPStatement();
                    };
                } catch (Exception E) {
                    logger.error(E);
                    continue;
                }
                docRecordSet.clear();
                break;
            }
        } catch (Exception E) {
            logger.error(E);

        }
    }

    private boolean updateData(List<Map<String, Object>> lists) {
        boolean retu = true;
        long st = System.currentTimeMillis();

        List<ResultSetFuture> future = new ArrayList<>();
        for (Object map : lists) {
            Map m = (Map) map;
            //绑定更新数据，并创建BoundStatement对象
            BoundStatement bs = update_statment.bind();
            setUpdateValue(m, bs);
            // 绑定更新记录的主�?

            columnMap.get(updatePrimaryKey).bind("key", m.get(updatePrimaryKey), bs);
            // 设置记录时间�?
            if (this.timestampKey.length() > 0 && m.containsKey(this.timestampKey)) {
                bs.setDefaultTimestamp((Long) m.get(this.timestampKey) * 1000000); // second to microsecond
            }
            future.add(session.executeAsync(bs));
        }
        // 处理执行结果
        int update = 0;
        for (int i = 0; i < future.size(); i++) {
            try {
                ResultSet rs = future.get(i).get();
                String result_string = rs.toString();
                if (result_string.contains("ERROR")) {
                    logger.error(result_string);
                } else if (rs.wasApplied()) {
//                    logger.info("update successfully");
                    //更新成功
                    update++;
                } else {
                    logger.error(result_string);
                    // TODO: 2017-03-25 此处可能是由于记录不存在等原因�?成更新未能生�?by linchan
                }

            } catch (Exception e) {
                logger.info(e);
                retu = false;
                // 异常捕捉�?30~234�?
            }
        }
        future.clear();
        logger.info("0st load " + lists.size() + "|expect: " + update + " records update to table " + tableSe2DBRule.getTableName() + " and mqName is " + tableSe2DBRule.getMqName() + " successfully using " + (System.currentTimeMillis() - st) / 1000000L + " ms ");
        return retu;
    }

    private String GetColumnName(String getMongodbName) throws IOException, InterruptedException {
        HashMap<String, String> h = RuntimeEnv.TABLE_CR_COLUMNS.get(tableSe2DBRule.getRuleName());

        if (h.get(getMongodbName) != null) {
            return h.get(getMongodbName);
        } else {
            return null;
        }
    }

    /**
     * 提交数据到Cassandra
     *
     * @param jsonList 数据列表
     * @return 返回是否提交成功
     * @throws IOException 抛出异常
     */
    private Type resolveType(DataType rawType) {
        switch (rawType.getName()) {
            case ASCII:
            case VARCHAR:
            case TEXT:
                return Type.STRING;
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Type.INT;
            case BIGINT:
            case VARINT:
                return Type.BIGINT;
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return Type.LIST;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + rawType);

        }
    }

    private enum Type {
        STRING {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else {
                    bs.setString(col, val.toString());
                }
            }
        }, TINYINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setByte(col, Long.valueOf(((String) val).toLowerCase()).byteValue());
                } else if (val instanceof Number) {
                    bs.setByte(col, ((Number) val).byteValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SMALLINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setShort(col, Long.valueOf(((String) val).toLowerCase()).shortValue());
                } else if (val instanceof Number) {
                    bs.setShort(col, ((Number) val).shortValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, INT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setInt(col, Long.valueOf(((String) val).toLowerCase()).intValue());
                } else if (val instanceof Number) {
                    bs.setInt(col, ((Number) val).intValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BIGINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
                } else if (val instanceof Number) {
                    bs.setLong(col, ((Number) val).longValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, FLOAT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setFloat(col, Long.valueOf(((String) val).toLowerCase()).floatValue());
                } else if (val instanceof Number) {
                    bs.setFloat(col, ((Number) val).floatValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, DOUBLE {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setDouble(col, Long.valueOf(((String) val).toLowerCase()).doubleValue());
                } else if (val instanceof Number) {
                    bs.setDouble(col, ((Number) val).doubleValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BLOB {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    try {
                        byte[] bytes = Hex.decodeHex(((String) val).toCharArray());
                        bs.setBytes(col, ByteBuffer.wrap(bytes));
                    } catch (DecoderException e) {
                        throw new IllegalArgumentException("wrong value of hex string, column: " + col);
                    }
                } else if (val instanceof byte[]) {
                    bs.setBytes(col, ByteBuffer.wrap((byte[]) val));
                } else if (val instanceof ByteBuffer) {
                    bs.setBytes(col, (ByteBuffer) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BOOLEAN {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setBool(col, Boolean.valueOf((String) val));
                } else if (val instanceof Boolean) {
                    bs.setBool(col, (Boolean) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, LIST {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof List) {
                    bs.setList(col, (List<?>) val);
                } else if (val instanceof Collection) {
                    List<?> l = new ArrayList<>((Collection<?>) val);
                    bs.setList(col, l);
                } else {
                    List<Object> l = new ArrayList<>();
                    l.add(val);
                    bs.setList(col, l);
                }
            }
        }, MAP {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Map) {
                    bs.setMap(col, (Map<?, ?>) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SET {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Set) {
                    bs.setSet(col, (Set<?>) val);
                } else if (val instanceof Collection) {
                    Set<?> s = new HashSet<>((Collection<?>) val);
                    bs.setSet(col, s);
                } else {
                    Set<Object> s = new HashSet<>();
                    s.add(val);
                    bs.setSet(col, s);
                }
            }
        };

        public void bind(String col, Object val, BoundStatement bs) {
            throw new IllegalArgumentException("wrong type of " + col);
        }
    }

    /**
     * 提交更新数据到Cassandra
     *
     * @param lists 数据列表
     */
    public void close() {
        try {
            if (session != null || cluster != null) {
                this.session.close();
                this.cluster.close();
            }
        } catch (Exception E) {
            logger.info(E);
        } finally {
            session = null;
            cluster = null;
            insert_statement = null;
            update_statment = null;
        }
    }

    /**
     * 映射Cassandra类型到{@linkplain Type}枚举
     *
     * @param rawType Cassandra类型对象
     * @return 返回Cassandra类型对应的Type枚举
     */
    public Session getSession() {
        return this.session;
    }
    //去重程序  调用去重服务

    private Object resolveType(Type tt, Object o) {
        switch (tt) {
            case STRING:
                return o.toString();
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Integer.valueOf(o.toString());
            case BIGINT:
                return Long.valueOf(o.toString());
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return o;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + tt);
        }
    }

    /**
     * 设置更新条件
     *
     * @return
     */
    private String getUpdateStr() {
        StringBuilder sb = new StringBuilder();
        // 遍历�?��更新的各个字段并写入where条件�?
        for (String updateField : updateFields) {
            sb.append(updateField)
                    .append("=:")
                    .append(updateField)
                    .append(",");
        }
        sb.deleteCharAt(sb.length() - 1);
        return sb.toString();
    }

    /**
     * 设置更新列的�?
     *
     * @param map 从消息队列中读取的数�?
     * @param bs
     */
    private void setUpdateValue(Map map, BoundStatement bs) {
        // 遍历�?��更新的各个列并分别设置更新�?
        for (String updateField : updateFields) {
            Type type = columnMap.get(updateField);
            Object val = map.get(updateField);
            logger.info(val);
            type.bind(updateField, val, bs);
        }
    }

    private String getUpdatesql(String querykey) {
        String[] querykeys = querykey.split(":");
        int length = querykey.split(":").length;
        String sql = "update " + tableName + " set "
                + getUpdateStr() + " where ";
        StringBuilder sb = new StringBuilder();
        sb.append(sql);
        int i = 0;
        for (String s : querykeys) {
            sb.append(s);
            i++;
            sb.append("=:" + "key" + i + " and ");
        }
        int j = sb.lastIndexOf(" and");
        String ss2 = sb.substring(0, j) + " if exists";
        return ss2;
    }

    private void setKey2(String querykey, Map m, BoundStatement bs) {
        int idx = 0;
        String[] querykeys = querykey.split(":");
        int i = 0;
        for (String s : querykeys) {
            i++;
            columnMap.get(s).bind("key" + i, m.get(s), bs);
        };
    }

}
